// -*-mode:c++; coding:utf-8-*-

#include <iostream>
#include <sstream>
#include <string>

using namespace std;

#include <boost/asio.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/thread/mutex.hpp>

#include <xbase_utility.hpp>
#include <naming_strategy.hpp>

#include "io_service_pool_impl.hpp"
#include "session_manager.hpp"
#include "parallel_scanner.hpp"

using namespace xbase;

typedef boost::system::error_code error_code;

unsigned scan_row_count = 0;

void on_get_next_row(row_result::ptr_type row_ptr, const error_code &ec)
{
	if(ec)
	{
		if (ec == xbase::error::eos)
		{
			std::cout << "scan finished, row count: " << scan_row_count <<endl;
		}
		else
		{
			std::cout << "get next row failed : " << ec
				<< " - " << ec.message() << std::endl;
		}
	}
	else
	{
		//cout << "row: {row_name: " << row_ptr->m_row_name << " columns: {";

		//row_result::cell_vector::iterator iter = row_ptr->m_cell_vector.begin();
		//while (iter != row_ptr->m_cell_vector.end())
		//{
		//	cell &c = *iter;
		//	unsigned version_count = c.get_version_count();

		//	std::cout << "{column_name: " << c.m_column_name;
		//	std::cout << " version_count: " << version_count << " versions: {";

		//	for (unsigned i=0;i<version_count;i++)
		//	{
		//		cell::version v;
		//		c.get_version(i, v);
		//		cout << "{" << v.first << "," << dump_string_hex(v.second) << "}, ";
		//	}
		//	cout << "}} ";

		//	++iter;
		//}
		//cout << "}}" << endl;

		//cout << "row: " << row_ptr->m_row_name << endl;

		++scan_row_count;

		if (scan_row_count%500 == 0)
		{
			cout << "scan_row_count: " << scan_row_count << endl;
		}
	}
}

int main(int argc, char **argv)
{
	if(argc < 7)
	{
		cout <<argv[0]<<" master_ip master_port table_name start_row end_row column_names"<< endl;
		return 0;
	}

	LOG_CONFIG("../../conf/log4cplus.test.conf");

	boost::asio::ip::tcp::endpoint endpoint(
		boost::asio::ip::address::from_string(argv[1]),
		::atoi(argv[2])
		);

	string table_name(argv[3]);
	string start_row(argv[4]);
	string end_row(argv[5]);
	vector<string> column_names;
	
	for (int i=6;i<argc;i++)
	{
		column_names.push_back(argv[i]);
	}

	//
	// 首先初始化service_pool并启动
	// 

	boost::shared_ptr<io_service_pool> pool(new impl::single_io_service());
	client_options co;
	pool->set_options(co);

	pool->run();

	session_manager sm(*pool);
	root_meta_cache rmc(*pool, endpoint, sm);
	location_cache lc(*pool, rmc, sm);

	cout<<" table_name: "<<table_name<<" start_row: "<<start_row<<" end_row: "<<end_row<<endl;

	parallel_scanner_ptr ps(new parallel_scanner(*pool, sm, lc, boost::bind(on_get_next_row, _1, _2), table_name, column_names, start_row, end_row));
	ps->start();

	// 结束
	std::string input;
	while(std::cin >> input)
	{
		if(input == "close")
		{
			sm.close();
		}
		else if(input == "quit")
		{
			break;
		}
	}

	pool->stop();
	pool->join();

	return 0;
}
